package io.debezium.connector.sqlserver.process;

import java.io.File;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import io.debezium.connector.process.BaseProcessCommitter;
import io.debezium.connector.process.BaseSourceProcessInfo;
import io.debezium.connector.sqlserver.SqlServerConnection;
import io.debezium.connector.sqlserver.SqlServerConnectorConfig;
import io.debezium.connector.sqlserver.sink.task.SqlServerSinkConnectorConfig;

public class SqlServerProcessCommitter extends BaseProcessCommitter {
    private static final Logger LOGGER = LoggerFactory.getLogger(io.debezium.connector.sqlserver.process.SqlServerProcessCommitter.class);
    private static final String FORWARD_SOURCE_PROCESS_PREFIX = "forward-source-process-";
    private static final String FORWARD_SINK_PROCESS_PREFIX = "forward-sink-process-";
    private static final String CREATE_COUNT_INFO_NAME = "start-event-index.txt";

    public static final Integer MAX_TABLE = 100;
    private static final String GET_CDC_TABLE = "select concat(\'cdc.\' , [capture_instance] , \'_CT\') as ct_name from cdc.change_tables";
    private static final String GET_CDC_DATA_PREFIX = "select [__$start_lsn],[__$seqval],[__$operation] from ";

    private List<String> changeTableList;
    private List<List> lsnLists;

    private final ThreadPoolExecutor threadPool = new ThreadPoolExecutor(1, 1, 100,
            TimeUnit.SECONDS, new LinkedBlockingQueue<>(1));
    private BaseSourceProcessInfo sourceProcessInfo;
    private SqlServerSinkProcessInfo sinkProcessInfo;
    private SqlServerConnection sqlServerConnection;
    private long createCount;
    private boolean isParallelBasedTransaction;

    public SqlServerProcessCommitter(String processFilePath, String prefix, int commitTimeInterval, int fileSizeLimit) {
        super(processFilePath, prefix, commitTimeInterval, fileSizeLimit);
    }

    public SqlServerProcessCommitter(SqlServerConnectorConfig connectorConfig, List<List> lsnLists,
                                     SqlServerConnection connection) {
        this(connectorConfig.filePath(), FORWARD_SOURCE_PROCESS_PREFIX,
                connectorConfig.commitTimeInterval(), connectorConfig.fileSizeLimit());
        this.fileFullPath = initFileFullPath(file + File.separator + FORWARD_SOURCE_PROCESS_PREFIX);
        this.currentFile = new File(fileFullPath);
        this.isAppendWrite = connectorConfig.appendWrite();
        deleteRedundantFiles(connectorConfig.filePath(),
                connectorConfig.processFileCountLimit(), connectorConfig.processFileTimeLimit());
        this.sqlServerConnection = connection;
        this.isParallelBasedTransaction = connectorConfig.shouldProvideTransactionMetadata();
        if (isParallelBasedTransaction) {
            this.sourceProcessInfo = BaseSourceProcessInfo.TRANSACTION_SOURCE_PROCESS_INFO;
            initLsnList(lsnLists);
        }
        else {
            this.sourceProcessInfo = BaseSourceProcessInfo.TABLE_SOURCE_PROCESS_INFO;
        }
        executeOutPutThread(connectorConfig.createCountInfoPath() + File.separator);
    }

    private void initLsnList(List<List> originLsnLists) {
        lsnLists = originLsnLists;
    }

    /**
     * Constructor
     *
     * @param connectorConfig SqlServerSinkConnectorConfig the connectorConfig
     */
    public SqlServerProcessCommitter(SqlServerSinkConnectorConfig connectorConfig) {
        this(connectorConfig.getSinkProcessFilePath(), FORWARD_SINK_PROCESS_PREFIX,
                connectorConfig.getCommitTimeInterval(), connectorConfig.getFileSizeLimit());
        this.fileFullPath = initFileFullPath(file + File.separator + FORWARD_SINK_PROCESS_PREFIX);
        this.currentFile = new File(fileFullPath);
        this.isAppendWrite = connectorConfig.isAppend();
        this.createCountInfoPath = connectorConfig.getCreateCountInfoPath();
        this.isParallelBasedTransaction = connectorConfig.isParallelBasedTransaction;
        sinkProcessInfo = SqlServerSinkProcessInfo.SINK_PROCESS_INFO;
        deleteRedundantFiles(connectorConfig.getSinkProcessFilePath(),
                connectorConfig.getProcessFileCountLimit(), connectorConfig.getProcessFileTimeLimit());
    }

    /**
     * Constructor
     *
     * @param failSqlPath String the fail sql path
     * @param fileSize int the file size
     */
    public SqlServerProcessCommitter(String failSqlPath, int fileSize) {
        super(failSqlPath, fileSize);
    }

    /**
     * statSourceProcessInfo
     *
     * @return SqlServerSourceProcessInfo
     */
    protected String statSourceProcessInfo() {
        long before = waitTimeInterval(true);
        sourceProcessInfo.setSpeed(before, commitTimeInterval);
        if (isParallelBasedTransaction) {
            BaseSourceProcessInfo processInfo = sourceProcessInfo.clone();
            refreshCreateCount();
            processInfo.setCreateCount(createCount);
            processInfo.setRest();
            processInfo.setTimestamp();
            return processInfo.toString();
        }
        sourceProcessInfo.setCreateCount(createCount);
        sourceProcessInfo.setRest();
        sourceProcessInfo.setTimestamp();
        return sourceProcessInfo.toString();
    }

    private void refreshCreateCount() {
        List<List> currentLsnLists = getCurrentLsnLists();
        long currentLsnListsSize = currentLsnLists.size();
        long lsnListsSize = lsnLists.size();
        for (int i = 0; i < currentLsnListsSize; i++) {
            if (i > lsnListsSize) {
                createCount += currentLsnLists.get(i).size();
            }
            else {
                createCount += currentLsnLists.get(i).size() - lsnLists.get(i).size();
            }
        }
        lsnLists = currentLsnLists;
    }

    protected String statSinkProcessInfo() {
        long before = waitTimeInterval(false);
        sinkProcessInfo.setSpeed(before, commitTimeInterval);
        sinkProcessInfo.setRest(sinkProcessInfo.getSkippedExcludeEventCount(), sinkProcessInfo.getSkippedCount());
        sinkProcessInfo.setTimestamp();
        long sourceCreateCount = inputCreateCount(createCountInfoPath + File.separator
                + CREATE_COUNT_INFO_NAME);
        while (sourceCreateCount != -1 && sourceCreateCount < sinkProcessInfo.getReplayedCount()
                + sinkProcessInfo.getSkippedCount() + sinkProcessInfo.getSkippedExcludeEventCount()) {
            sourceCreateCount = inputCreateCount(createCountInfoPath + File.separator
                    + CREATE_COUNT_INFO_NAME);
        }
        if (sourceCreateCount != -1) {
            sinkProcessInfo.setOverallPipe(sourceCreateCount);
        }
        return sinkProcessInfo.toString();
    }

    /*
     * return all ct_table names
     */
    public List<String> getChangeTableList() {
        return changeTableList;
    }

    private List<String> getCurrentChangeTableList() {
        List<String> currentChangeTableList = new ArrayList<>();
        try {
            sqlServerConnection.query(GET_CDC_TABLE, rs -> {
                while (rs.next()) {
                    currentChangeTableList.add(rs.getString("ct_name"));
                }
            });
        }
        catch (SQLException e) {
            LOGGER.error("SQL exception occurred when query the current event index.");
        }

        System.out.println(currentChangeTableList);
        changeTableList = currentChangeTableList;
        return currentChangeTableList;
    }

    private List<List> getCurrentLsnLists() {
        List<List> currentLsnLists = new ArrayList<List>();
        List<String> currentChangeTableList = getCurrentChangeTableList();

        for (String tableName : currentChangeTableList) {
            try {
                sqlServerConnection.query(GET_CDC_DATA_PREFIX + tableName, rs2 -> {
                    List<HashMap> currentLsnList = new ArrayList<HashMap>();
                    while (rs2.next()) {
                        HashMap<String, String> modelMap = new HashMap<>();
                        modelMap.put("change_lsn", rs2.getString(2));
                        modelMap.put("commit_lsn", rs2.getString(1));
                        modelMap.put("operation", rs2.getString(3));
                        currentLsnList.add(modelMap);
                    }
                    System.out.println(currentLsnList);
                    currentLsnLists.add(currentLsnList);
                });
            }
            catch (SQLException e) {
                LOGGER.error("SQL exception occurred when query the current event index.");
            }

        }
        return currentLsnLists;
    }

    private long waitTimeInterval(boolean isSource) {
        long before;
        if (isSource) {
            before = sourceProcessInfo.getPollCount();
        }
        else {
            before = sinkProcessInfo.getReplayedCount();
        }
        try {
            Thread.sleep(commitTimeInterval * 1000L);
        }
        catch (InterruptedException exp) {
            LOGGER.warn("Interrupted exception occurred", exp);
        }
        return before;
    }

    private void executeOutPutThread(String dirPath) {
        threadPool.execute(() -> {
            if (!initFile(dirPath).exists()) {
                LOGGER.warn("Failed to output source create count, the sink overallPipe will always be 0.");
                threadPool.shutdown();
                return;
            }
            while (true) {
                try {
                    Thread.sleep(1000);
                }
                catch (InterruptedException exp) {
                    LOGGER.error("Interrupted exception occurred while thread sleeping", exp);
                }
                outputCreateCountInfo(dirPath + CREATE_COUNT_INFO_NAME, sourceProcessInfo
                        .getCreateCount() - sourceProcessInfo.getSkippedExcludeCount());
            }
        });
    }

}
